package com.ydl.learning.flink.demo.broadcast

import com.ydl.learning.flink.demo.Utils
import com.ydl.learning.flink.demo.mySink.LogSink
import com.ydl.learning.flink.demo.mySource.StringSource
import org.apache.flink.api.common.state.MapStateDescriptor
import org.apache.flink.api.common.typeinfo.BasicTypeInfo
import org.apache.flink.api.scala._
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction
import org.apache.flink.util.Collector
import org.slf4j.{Logger, LoggerFactory}

/**
 * 广播流demo
 * 广播流分keyed流和non-keyed流，本例是non-keyed流
 * @author ydl
 * @since 2021/5/13
 */
object BroadcastStreamDemo extends App with Utils {
  private val log: Logger = LoggerFactory.getLogger("BroadcastStreamDemo")
  val info = getSocketTextStream(9999)
  val broadcastStream = info._1.addSource(new StringSource).setParallelism(1).name("broadcastStream")
  //广播描述，定义广播的数据的名称，key类型，value类型，keyTypeInfo在keyed流里有用，此处没啥作用
  val broadcastStateDescriptor: MapStateDescriptor[String, String] = new MapStateDescriptor[String, String](
    "broadcast",
    BasicTypeInfo.STRING_TYPE_INFO,
    BasicTypeInfo.STRING_TYPE_INFO
  )
  val broadcastConfig = broadcastStream.broadcast(broadcastStateDescriptor)

  val stream = info._2.connect(broadcastConfig).process(new BroadcastProcessFunction[String, String, String] {
    var element: String = _

    /**
     * 接受到数据流的每条消息时会调用
     *
     * @param value
     * @param ctx
     * @param out
     */
    override def processElement(value: String,
                                ctx: BroadcastProcessFunction[String, String, String]#ReadOnlyContext,
                                out: Collector[String]): Unit = {
      log.info("concat value:{} {}", List(value, element): _*)
      out.collect(value + "|" + element)
    }

    /**
     * 每次收到广播流的记录时会调用
     *
     * @param value
     * @param ctx
     * @param out
     */
    override def processBroadcastElement(value: String,
                                         ctx: BroadcastProcessFunction[String, String, String]#Context,
                                         out: Collector[String]): Unit = {
      log.info("receive broadcast:{}", value)
      element = value
    }
  })
  stream.addSink(new LogSink[String]).name("log")
  info._1.execute("BroadcastStreamDemo")
}
